package com.shujia.flink.state

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util
import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object Demo1ListState {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val carDS: DataStream[String] = env.socketTextStream("master", 8888)

    //解析数据
    val cardAndSpeedDS: DataStream[(Long, Double)] = carDS.map(line => {
      val jsonObj: JSONObject = JSON.parseObject(line)
      val card: Long = jsonObj.getLong("card")
      val speed: Double = jsonObj.getDouble("speed")
      (card, speed)
    })
    /**
     * 实时计算每隔kafkou的平均车速
     */
    //安装卡口分组
    val keyByDS: KeyedStream[(Long, Double), Long] = cardAndSpeedDS.keyBy(_._1)

    val avgSpeedDS: DataStream[(Long, Double)] = keyByDS
      .process(new KeyedProcessFunction[Long, (Long, Double), (Long, Double)] {

        /**
         * ListState: 集合状态
         * 和scala普通集合的区别
         * 1、为每一个key再状态中保存一个结果
         * 2、flink的状态会定时被checkpoint保存到hdfs中，任务执行失败，可以基于hdfs中的数据恢复任务，保证结果不丢失
         *
         * checkpoint需要手动开启
         */
        var speedsState: ListState[Double] = _

        //open方法再task启动的时候执行，一班用于初始化，比如创建数据库连接或者初始化状态
        override def open(parameters: Configuration): Unit = {
          //获取flin运行环境对象
          val context: RuntimeContext = getRuntimeContext
          //创建状态描述对象
          val listStateDesc = new ListStateDescriptor[Double]("speeds", classOf[Double])

          /**
           * ListState: 为每一个key保存一个集合状态，可以当成一个普通集合使用
           */
          //初始化状态
          speedsState = context.getListState(listStateDesc)
        }

        /**
         * processElement: 每一条数据执行一次
         *
         * @param value ：一行数据
         * @param ctx   ： 上下文对象
         * @param out   ： 用于将数据发送到下游
         */
        override def processElement(value: (Long, Double),
                                    ctx: KeyedProcessFunction[Long, (Long, Double), (Long, Double)]#Context,
                                    out: Collector[(Long, Double)]): Unit = {

          val (card: Long, speed: Double) = value

          //将车速保存到状态中
          speedsState.add(speed)

          //计算平均车速
          val iter: util.Iterator[Double] = speedsState.get().iterator()

          //导入将java集合转换成scala集合的隐式转换
          import scala.collection.JavaConverters._

          val speeds: List[Double] = iter.asScala.toList

          val avgSpeed: Double = speeds.sum / speeds.length

          //将数据发送到下游
          out.collect((card, avgSpeed))
        }
      })

    avgSpeedDS.print()

    env.execute()
  }

}
